package com.zhss.dfs.server;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.zhss.dfs.server.model.FileOperate;
import com.zhss.dfs.server.model.FileReplicaTask;
import com.zhss.rpc.model.NameNodeModel;
import com.zhss.rpc.service.NameNodeServiceGrpc;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;

/**
 * NameNode的rpc服务的接口
 *
 * @author zhonghuashishan
 */
public class NameNodeRpcServer {

    private final int port;

    private final Server server;

    public static final Integer STATUS_SUCCESS = 1;
    public static final Integer STATUS_FAILURE = 2;
    public static final Integer STATUS_SHUTDOWN = 3;
    public static final Integer NAME_REPEAT = 4;
    public static final Integer RESTART = 5;


    /**
     * 负责管理元数据的核心组件
     */
    private FSNamesystem namesystem;
    /**
     * 负责管理集群中所有的datanode的组件
     */
    private DataNodeManager datanodeManager;

    public NameNodeRpcServer(
            FSNamesystem namesystem,
            DataNodeManager datanodeManager) {
        this.namesystem = namesystem;
        this.datanodeManager = datanodeManager;
        this.port = 50072;
        this.server = ServerBuilder.forPort(port)
                .addService(new NameNodeServiceImpl(this.datanodeManager, this.namesystem))
                .build();
    }


    // 启动服务器的方法
    public void start() throws Exception {
        server.start();
        System.out.println("服务器已启动，端口号：" + port);
        // 注册 JVM 关闭钩子，用于在 JVM 关闭时停止服务器
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            // this.namesystem.
            System.out.println("正在关闭服务器");
            this.stop();
        }));
    }

    // 停止服务器的方法
    private void stop() {
        if (server != null) {
            server.shutdown();
        }
    }

    public Server getServer() {
        return server;
    }

    // 内部类，实现 gRPC 服务接口
    private static class NameNodeServiceImpl extends NameNodeServiceGrpc.NameNodeServiceImplBase {

        private DataNodeManager dataNodeManager;

        private FSNamesystem fsNamesystem;

        private Boolean isRunning = true;

        public NameNodeServiceImpl(DataNodeManager dataNodeManager, FSNamesystem fsNamesystem) {
            this.dataNodeManager = dataNodeManager;
            this.fsNamesystem = fsNamesystem;
        }

        /**
         * dataNode进行注册
         *
         * @param request
         * @param responseObserver
         */
        @Override
        public void register(NameNodeModel.RegisterRequest request, StreamObserver<NameNodeModel.RegisterResponse> responseObserver) {
            String ip = request.getIp();
            String hostname = request.getHostname();
            int port = request.getPort();
            Boolean registerSuccess = dataNodeManager.register(ip, hostname, port);
            NameNodeModel.RegisterResponse response = null;
            if (registerSuccess) {
                System.out.println(ip + " " + hostname + "  已经注册");
                response = NameNodeModel.RegisterResponse.newBuilder()
                        .setStatus(STATUS_SUCCESS)
                        .build();
            } else {
                response = NameNodeModel.RegisterResponse.newBuilder()
                        .setStatus(STATUS_FAILURE)
                        .build();
            }
            // 发送回复消息给客户端
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        }


        /**
         * dataNode 进行心跳检测
         *
         * @param request
         * @param responseObserver
         */
        @Override
        public void heartbeat(NameNodeModel.HeartbeatRequest request, StreamObserver<NameNodeModel.HeartbeatResponse> responseObserver) {
            String ip = request.getIp();
            String hostname = request.getHostname();
            Boolean result = dataNodeManager.heartBeat(ip, hostname);
            DataNodeInfo dataNodeInfo = dataNodeManager.getDataNode(hostname, ip);
            FileReplicaTask fileReplicaTask = null;
            JSONArray jsonArray = new JSONArray();
            // 获取需要进行拷贝的任务
            while ((fileReplicaTask = dataNodeInfo.getFileReplicaTask() )!= null) {
                jsonArray.add(JSONObject.toJSONString(fileReplicaTask));
            }
            NameNodeModel.HeartbeatResponse response = null;
            if (!result) {
                response = NameNodeModel.HeartbeatResponse.newBuilder()
                        .setStatus(RESTART)
                        .build();
                System.out.println("此时nameNode重启，dataNode数据已经不存在");
            } else {
                System.out.println(ip + " " + hostname + "  进行心跳检测");
                response = NameNodeModel.HeartbeatResponse.newBuilder()
                        .setStatus(STATUS_SUCCESS)
                        .build();
            }
            // 发送回复消息给客户端
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        }


        /**
         * 创建目录
         *
         * @param request
         * @param responseObserver
         */
        @Override
        public void mkdir(NameNodeModel.MkdirRequest request, StreamObserver<NameNodeModel.MkdirResponse> responseObserver) {
            try {
                if (!isRunning) {
                    // 停止
                    NameNodeModel.MkdirResponse mkdirResponse = NameNodeModel.MkdirResponse.newBuilder()
                            .setStatus(STATUS_SHUTDOWN)
                            .setMessage("成功停止")
                            .build();
                    responseObserver.onNext(mkdirResponse);
                    responseObserver.onCompleted();
                    return;
                }
                FileOperate operate = new FileOperate("MKDIR", request.getPath());
                this.fsNamesystem.mkdir(operate);
				System.out.println("创建目录： " + request.getPath());

                NameNodeModel.MkdirResponse mkdirResponse = NameNodeModel.MkdirResponse.newBuilder()
                        .setStatus(STATUS_SUCCESS)
                        .setMessage("创建成功")
                        .build();
                responseObserver.onNext(mkdirResponse);
                responseObserver.onCompleted();
            } catch (Exception e) {
                e.printStackTrace();
				NameNodeModel.MkdirResponse mkdirResponse = NameNodeModel.MkdirResponse.newBuilder()
						.setStatus(STATUS_FAILURE)
						.setMessage(e.getMessage())
						.build();
				responseObserver.onNext(mkdirResponse);
				responseObserver.onCompleted();
            }
        }

        /**
         *  backupNode 定时拉取日志
         * @param request
         * @param responseObserver
         */
        @Override
        public void fetchLog(NameNodeModel.FetchLogRequest request, StreamObserver<NameNodeModel.FetchLogResponse> responseObserver) {
            if (!isRunning) {
                NameNodeModel.FetchLogResponse fetchLogResponse = NameNodeModel.FetchLogResponse.newBuilder()
                        .setData("[]")
                        .setStatus(STATUS_SHUTDOWN)
                        .build();
                responseObserver.onNext(fetchLogResponse);
                responseObserver.onCompleted();
                return;
            }
            int size = request.getSize();
            long syncTxid = request.getSyncTxid();
            JSONArray jsonArray = this.fsNamesystem.fetchLogData(size,syncTxid);
            NameNodeModel.FetchLogResponse fetchLogResponse = NameNodeModel.FetchLogResponse.newBuilder()
                    .setData(jsonArray == null ? "[]" : jsonArray.toJSONString())
                    .setStatus(STATUS_SUCCESS).build();
            responseObserver.onNext(fetchLogResponse);
            responseObserver.onCompleted();
        }

        @Override
        public void updateCheckPonterTxid(NameNodeModel.UpdateTxidRequest request, StreamObserver<NameNodeModel.UpdateTxidResponse> responseObserver) {
            long checkPointerTxid = request.getTxid();
            this.fsNamesystem.setCheckPointerTxid(checkPointerTxid);
            this.fsNamesystem.deleteLogByTxid();
            NameNodeModel.UpdateTxidResponse updateTxidResponse = NameNodeModel.UpdateTxidResponse.newBuilder()
                    .setStatus(STATUS_SUCCESS).build();
            responseObserver.onNext(updateTxidResponse);
            responseObserver.onCompleted();
        }

        @Override
        public void shutDown(NameNodeModel.ShutDownRequest request, StreamObserver<NameNodeModel.ShutDownResponse> responseObserver) {
            isRunning = false;
            fsNamesystem.flush();
            fsNamesystem.saveCheckPinterTxid();
        }

        @Override
        public void createFile(NameNodeModel.CreateFileRequest request, StreamObserver<NameNodeModel.CreateFileResponse> responseObserver) {
            if (!isRunning) {
                NameNodeModel.CreateFileResponse createFileResponse = NameNodeModel.CreateFileResponse.newBuilder()
                        .setStatus(STATUS_FAILURE).setMsg("服务已经关闭").build();
                responseObserver.onNext(createFileResponse);
                responseObserver.onCompleted();
                return;
            }
            String fileName = request.getFileName();
            FileOperate fileOperate = new FileOperate("CREATE", fileName);
            Boolean success = fsNamesystem.createFile(fileOperate);
            if (!success) {
                NameNodeModel.CreateFileResponse createFileResponse = NameNodeModel.CreateFileResponse.newBuilder()
                        .setStatus(NAME_REPEAT).setMsg("文件名重复").build();
                responseObserver.onNext(createFileResponse);
                responseObserver.onCompleted();
                return;
            }
            NameNodeModel.CreateFileResponse createFileResponse = NameNodeModel.CreateFileResponse.newBuilder()
                    .setStatus(STATUS_SUCCESS).build();
            responseObserver.onNext(createFileResponse);
            responseObserver.onCompleted();
        }

        @Override
        public void allocateDataNode(NameNodeModel.AllocateDataNodeRequest request, StreamObserver<NameNodeModel.AllocateDataNodeResponse> responseObserver) {
            long fileSize = request.getFileSize();
            String dataNodes = dataNodeManager.getAllocateDataNode(fileSize);
            NameNodeModel.AllocateDataNodeResponse allocateDataNodeResponse = NameNodeModel.AllocateDataNodeResponse.newBuilder()
                    .setAllocateDateNodes(dataNodes)
                    .build();
            responseObserver.onNext(allocateDataNodeResponse);
            responseObserver.onCompleted();
        }

        @Override
        public void informReceivedReplicate(NameNodeModel.InformReceivedReplicateRequest request, StreamObserver<NameNodeModel.InformReceivedReplicateResponse> responseObserver) {
            String fileName = request.getFileName();
            String hostname = request.getHostname();
            String ip = request.getIp();
            long fileSize = request.getFileSize();
            fsNamesystem.informReceivedReplicate(fileName,hostname,ip,fileSize);
            NameNodeModel.InformReceivedReplicateResponse response = NameNodeModel.InformReceivedReplicateResponse.newBuilder()
                    .setStatus(STATUS_SUCCESS)
                    .build();
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        }

        @Override
        public void reportCompleteFileInfo(NameNodeModel.ReportCompleteFileRequest request,
                                           StreamObserver<NameNodeModel.ReportCompleteFileResponse> responseObserver) {
            String hostname = request.getHostname();
            String ip = request.getIp();
            String filenames = request.getFilenames();
            long fileStoreSize = request.getFileStoreSize();
            fsNamesystem.reportCompleteFileInfo(hostname,ip,filenames,fileStoreSize);
            DataNodeInfo dataNode = dataNodeManager.getDataNode(hostname, ip);
            JSONArray jsonArray = new JSONArray();
            FileReplicaTask fileReplicaTask = null;
            while ((fileReplicaTask = dataNode.getDelFileReplicaTask()) != null) {
                jsonArray.add(JSONObject.toJSONString(fileReplicaTask));
            }
            NameNodeModel.ReportCompleteFileResponse response = NameNodeModel.ReportCompleteFileResponse.newBuilder()
                    .setStatus(STATUS_SUCCESS).build();
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        }

        @Override
        public void getDataNodeForFile(NameNodeModel.GetDataForFileRequest request, StreamObserver<NameNodeModel.GetDataForFileResponse> responseObserver) {
            String fileName = request.getFileName();
            String dataNodeInfo = fsNamesystem.getDataNodeForFile(fileName);
            NameNodeModel.GetDataForFileResponse response = NameNodeModel.GetDataForFileResponse.newBuilder().setDataNodeInfo(dataNodeInfo).setStatus(STATUS_SUCCESS).build();
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        }


        @Override
        public void reallocateDataNode(NameNodeModel.ReallocateDataNodeRequest request, StreamObserver<NameNodeModel.ReallocateDataNodeResponse> responseObserver) {
            String excludeDataNode = request.getExcludeDataNode();
            long fileSize = request.getFileSize();
            String dataNode = fsNamesystem.reallocateDataNode(fileSize,excludeDataNode);
            NameNodeModel.ReallocateDataNodeResponse response = NameNodeModel.ReallocateDataNodeResponse.newBuilder()
                    .setDataNode(dataNode)
                    .setStatus(STATUS_SUCCESS)
                    .build();
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        }

        @Override
        public void chooseOtherDataNode(NameNodeModel.ChooseOtherDataNodeRequest request, StreamObserver<NameNodeModel.ChooseOtherDataNodeResponse> responseObserver) {
            String excludeDataNode = request.getExcludeDataNode();
            String fileName = request.getFileName();
            String dataNode = fsNamesystem.chooseOtherDataNode(excludeDataNode, fileName);
            NameNodeModel.ChooseOtherDataNodeResponse response = NameNodeModel.ChooseOtherDataNodeResponse.newBuilder()
                    .setDataNode(dataNode)
                    .setStatus(STATUS_SUCCESS).build();
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        }

        @Override
        public void rebalance(NameNodeModel.RebalanceRequest request, StreamObserver<NameNodeModel.RebalanceResponse> responseObserver) {
            fsNamesystem.rebalace();
            NameNodeModel.RebalanceResponse response = NameNodeModel.RebalanceResponse.newBuilder()
                    .setStatus(STATUS_SUCCESS)
                    .build();
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        }
    }




}
